package com.uw.uwrpc.core.registry.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.uw.uwrpc.core.api.RegistryCenter;
import com.uw.uwrpc.core.config.UWRegistryProperty;
import com.uw.uwrpc.core.meta.InstanceMeta;
import com.uw.uwrpc.core.meta.ServiceMeta;
import com.uw.uwrpc.core.registry.ChangedListener;
import com.uw.uwrpc.core.registry.RegistryCenterChangedEvent;
import com.uw.uwrpc.core.util.HttpUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class UWRegistryCenter implements RegistryCenter {

    private final UWRegistryProperty uwRegistryProperty;

    private ScheduledExecutorService consumerHealthCheckSchedule;
    private ScheduledExecutorService providerHealthCheckSchedule;

    private final MultiValueMap<InstanceMeta, ServiceMeta> heartbeatMap = new LinkedMultiValueMap<>();
    private final Map<String, Long> serviceVersionMap = new HashMap<>();


    public UWRegistryCenter(UWRegistryProperty uwRegistryProperty) {
        this.uwRegistryProperty = uwRegistryProperty;
    }


    @Override
    public void start() {
        log.debug(" ====> UwRegistryCenter start with server {}", uwRegistryProperty.getServers());

        consumerHealthCheckSchedule = Executors.newScheduledThreadPool(1);
        providerHealthCheckSchedule = Executors.newScheduledThreadPool(1);

        providerHealthCheckSchedule.scheduleWithFixedDelay(() -> {
            for (Map.Entry<InstanceMeta, List<ServiceMeta>> entry : heartbeatMap.entrySet()) {
                InstanceMeta instanceMeta = entry.getKey();
                List<ServiceMeta> serviceMetaList = entry.getValue();

                String services = serviceMetaList.stream().map(ServiceMeta::serviceSign).collect(Collectors.joining(","));
                Long timestamp = HttpUtil.httpPost(JSON.toJSONString(instanceMeta), uwRegistryProperty.getServers() + "/heartbeats?services=" + services, Long.class);
                log.debug(" ====> provider heartbeat timestamp:{}", timestamp);
            }

        }, 5, 5, TimeUnit.SECONDS);
    }

    @Override
    public void stop() {
        gracefulShutdown(consumerHealthCheckSchedule);
        gracefulShutdown(providerHealthCheckSchedule);
        log.debug(" ====> UwRegistryCenter stop with server {}", uwRegistryProperty.getServers());
    }

    @Override
    @SneakyThrows
    public void register(ServiceMeta serviceMeta, InstanceMeta instanceMeta) {
        log.debug(" ====> register instance:{} for {}", instanceMeta, serviceMeta);
        InstanceMeta result = HttpUtil.httpPost(JSON.toJSONString(instanceMeta), uwRegistryProperty.getServers() + "/register?service=" + serviceMeta.serviceSign(), InstanceMeta.class);
        heartbeatMap.add(instanceMeta, serviceMeta);
        log.debug(" ====> register instance result:{}", result);
    }

    @Override
    @SneakyThrows
    public void unregister(ServiceMeta serviceMeta, InstanceMeta instanceMeta) {
        log.debug(" ====> unregister instance:{} for {}", instanceMeta, serviceMeta);
        InstanceMeta result = HttpUtil.httpPost(JSON.toJSONString(instanceMeta), uwRegistryProperty.getServers() + "/unregister?service=" + serviceMeta.serviceSign(), InstanceMeta.class);
        heartbeatMap.remove(instanceMeta, serviceMeta);
        log.debug(" ====> unregister instance result:{}", result);
    }

    @Override
    @SneakyThrows
    public List<InstanceMeta> fetchAll(ServiceMeta serviceMeta) {
        log.debug(" ====> fetchAll instance for {}", serviceMeta);
        List<InstanceMeta> result = HttpUtil.httpGet(uwRegistryProperty.getServers() + "/fetchAll?service=" + serviceMeta.serviceSign(), new TypeReference<>() {
        });
        log.debug(" ====> fetchAll instance result:{}", result);
        return result;
    }

    @Override
    public void subscribe(ServiceMeta serviceMeta, ChangedListener listener) {
        consumerHealthCheckSchedule.scheduleWithFixedDelay(() -> {
            Long curVersion = serviceVersionMap.getOrDefault(serviceMeta.serviceSign(), -1L);
            Long newVersion = HttpUtil.httpGet(uwRegistryProperty.getServers() + "/version?service=" + serviceMeta.serviceSign(), Long.class);
            log.debug(" ====> curVersion:{}, newVersion:{}", curVersion, newVersion);
            if (curVersion < newVersion) {
                List<InstanceMeta> instances = fetchAll(serviceMeta);
                listener.onChange(new RegistryCenterChangedEvent(instances));
                serviceVersionMap.put(serviceMeta.serviceSign(), newVersion);
            }
        }, 1, 5, TimeUnit.SECONDS);

    }

    private void gracefulShutdown(ScheduledExecutorService executorService) {
        executorService.shutdown();
        try {
            executorService.awaitTermination(1000, TimeUnit.MILLISECONDS);
            if (!executorService.isTerminated()) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            // ignore
        }
    }
}
